package cn.doitedu.rules.service;

import cn.doitedu.date.gen.LogBean;
import cn.doitedu.rules.dao.UserHistoryEventClickHouseDaoImpl;
import cn.doitedu.rules.dao.UserHistoryEventDao;
import cn.doitedu.rules.dao.UserProfileDao;
import cn.doitedu.rules.dao.UserProfileHbaseDaoImpl;
import cn.doitedu.rules.pojo.EventParam;
import cn.doitedu.rules.pojo.RuleCondition;
import cn.doitedu.rules.utils.EventComparator;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.Set;
/*
* 自定义查询服务   用来开启连接
* */
public class QueryService {


    private UserProfileDao userProfileDao;
    private UserHistoryEventDao userHistoryEventDao;
    private Connection hbaseConn;
    private java.sql.Connection clickHouseConn;
    private ParameterTool parameters;

    public QueryService(ParameterTool parameters) {
        this.parameters = parameters;
        try {
            //创建Hbase连接
            org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
            hbaseConf.set("hbase.zookeeper.quorum", parameters.getRequired("hbase.url"));
            hbaseConf.set("hbase.zookeeper.property.clientPort", parameters.getRequired("hbase.port"));
            hbaseConn = ConnectionFactory.createConnection(hbaseConf);
            userProfileDao = new UserProfileHbaseDaoImpl(hbaseConn);

            //创建ClickHouse连接
            clickHouseConn = DriverManager.getConnection(parameters.getRequired("clickhouse.url"));
            userHistoryEventDao = new UserHistoryEventClickHouseDaoImpl(clickHouseConn);
        } catch (IOException e1) {
            throw new RuntimeException("建立Hbase连接出现异常", e1);
        } catch (SQLException e2) {
            throw new RuntimeException("建立ClickHouse连接持续异常", e2);
        }

    }


    public void close() throws Exception {
        hbaseConn.close();
        clickHouseConn.close();
    }

    public boolean isMatch(LogBean event, RuleCondition condition) throws Exception {
        //LogBean event用户当前的行为数据
        //RuleCondition condition 事先预设好的规则

        //1.触发的行为：eventId：K，属性： p2 = v1 （0个或多个属性）
        EventParam triggerEvent = condition.getTriggerEvent();
        boolean isTrigger = EventComparator.compare(event, triggerEvent);
        if (!isTrigger) {
            return false;
        }

        //2.用户画像条件：deviceId： 000019  画像标签：tag11 = v5， tag89 = v1, tag99 = v2
        Map<String, String> userProfileConditions = condition.getUserProfileCondition();
        boolean isMatch = userProfileDao.compareTags(event, userProfileConditions);
        if(!isMatch) {
            return false;
        }

        //3.历史行为数据: SQL, 时间段，属性properties['p1'] = v1, properties['p2'] = v3
        List<EventParam> userActionCountConditions = condition.getUserActionCountCondition();
        boolean isMatchCount = userHistoryEventDao.isMatchCount(event, userActionCountConditions);
        if (!isMatchCount) {
            return false;
        }

        return true;
    }
}
